/**
 * Copyright (c) 2023 murenchao
 * taomu is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *       http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */
package cool.taomu.thread

import cool.taomu.lang.TaomuRuntime
import cool.taomu.utils.ThriftUtils
import cool.taomu.utils.h2.H2Utils
import java.io.Serializable
import java.nio.ByteBuffer
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
import org.apache.commons.lang3.SerializationUtils
import org.apache.thrift.TException
import org.eclipse.xtend.lib.annotations.Accessors
import org.slf4j.LoggerFactory

class NetworkThreadClient {

	@Accessors
	static class JvmInfo {
		String host;
		int port;
	}

	def static void execute(String host, Serializable ser) throws Exception{
		val serbyte = SerializationUtils.serialize(ser);
		var h2 = new H2Utils();
		var sql = "select host,port from jvminfo where full_load = 0 and host != ? order by cpu_core desc limit 1"
		var res = h2.select(sql, JvmInfo, #[host]);
		if (!res.nullOrEmpty) {
			var _res = res.get(0);
			var thost = _res.host
			var tport = _res.port
			ThriftUtils.rpc(thost, tport, [
				var NetworkThread.Client client = new NetworkThread.Client(it);
				client.execute(ByteBuffer.wrap(serbyte));
			]);
		} else {
			throw new NullPointerException("集群列表为空了");
		}
	}
}

class TaomuRunsPolicy extends CallerRunsPolicy {
	val static LOG = LoggerFactory.getLogger(TaomuRunsPolicy);
	val String host;

	new(String host) {
		this.host = host;
	}

	@Accessors
	static class JvmInfo {
		String host;
		int port;
	}

	def dispatch rejectedExecution(TaomuRuntime r, ThreadPoolExecutor executor) {
		LOG.info("TaomuRuntime当前被拒绝任务为:{}", r.toString());
		try {
			NetworkThreadClient.execute(host, r.data);
		} catch (TException e) {
			LOG.info("网络异常：", e);
			super.rejectedExecution(r, executor);
		}
	}

	def dispatch rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		LOG.info("当前被拒绝任务为:{}", r.toString());
		super.rejectedExecution(r, executor);
	}

}
